package com.atguigu.state;

import com.atguigu.pojo.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import util.SourceUtil;

import java.time.Duration;

/*
* @date 2024/11/22 10:03
*
* 网站中一个非常经典的例子，就是统计一段时间内热门url
*
* 例如：需要统计最近10秒钟内最热门的两个url链接，并且每5秒钟更新一次
* 关键词，统计时间间隔最近10s，每5秒钟更新一次：使用滑动窗口。热门度：通过访问量表示
*
* 使用全窗口函数实现，将窗口中的数据收集起来，等到窗口触发计算时，对收集的数据进行统计（统计每个url的点击次数），
* 再对窗口统计的结果进行排序，输出前N名
*
* */

public class Flink12_TopN_ProcessAllWindowFunction {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //数据生成器模拟生成数据，设置水位线
        SingleOutputStreamOperator<Event> ds = env.fromSource(SourceUtil.getSource(), WatermarkStrategy.noWatermarks(), "DataGenSource")
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );

        ds.print("INPUT");


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
